/***************************************************************************************
  *Copyright(C),2010-2016,Sumscope
  *FileName   : SortBuffer.cpp
  *Author     : scofined.qi
  *Version    : 1.0
  *Date       : 2016/06/03
  *Desc       : //用于说明此程序文件完成的主要功能
  *Relation   :
  *Other      : //其他内容说明
  *Function   : //主要函数列表，每条记录应包含函数名及功能简要说明
  *History    : //修改历史纪录列表，每条修改记录应包含修改日期，修改者以及修改内容简介
  ***************************************************************************************/
#include "SortBuffer.h"

#ifdef WIN32
#define BASSERT assert
#else
#define BASSERT //
#endif

namespace qb{
	namespace ht{
		SortBuffer::SortBuffer() :m_all_columns(false)
		{

		}

		SortBuffer::~SortBuffer()
		{
			if (m_buffer){ delete[] m_buffer; m_buffer = 0; }
		}

		void SortBuffer::CLearColumns()
		{
			m_columns.clear();

			// 最开始的4个字节用于保存记录号
			// 后续
			m_record_length = 0;
		}

		void SortBuffer::AddColumn(const ColumnInfo& ci, bool dec)
		{
			m_columns.push_back(ci);
			ColumnInfo& last = m_columns.at(m_columns.size() - 1);
			last.SetSortDecFlag(dec);
			if (!m_all_columns)
			{
				last.SetOffset(m_record_length);
				m_record_length += ci.GetFieldBytes();
			}
		}

		char* SortBuffer::PrepareBuffer(int rowCount)
		{
			// m_record_length 可按照4字节边界对齐，使得性能更好
			if (m_record_length % 4 != 0)
				m_record_length = (m_record_length / 4 + 1) * 4;
			else
				m_record_length += 4;

			m_record_count = rowCount;
			int total = rowCount*m_record_length;
			if (m_buffer_size < total || m_buffer == NULL)
			{
				if (m_buffer)
					delete[] m_buffer;
				m_buffer = new char[total];
				m_buffer_size = total;
				memset(m_buffer, 0, total);
			}
			memset(m_buffer, 0, m_buffer_size);
			return m_buffer;
		}

		int SortBuffer::GetRecordId(int row)const
		{
			// record在每个记录的最开始4个字节
            BASSERT(row >= 0 && row < m_record_count);
            BASSERT(m_buffer != 0);

			int offset = row*m_record_length + m_columns.at(m_columns.size() - 1).GetOffset();
			return *((int*)(m_buffer + offset));
		}

        Range SortBuffer::FindRow(RecordId rid, const void* record, int size, int rowCount, const std::vector<ColumnInfo>& columns)
		{
			// 抽取rid和排序字段构成的记录
            BASSERT(rid >= 0 && record);

            char* sort_record = (char*)alloca(m_record_length + 1);
			memset(sort_record, 0, m_record_length + 1);
			*((int*)sort_record) = rid;
			char* ptr = sort_record + sizeof(rid);
			for (int i = 0; i < (int)m_columns.size(); i++)
			{
				const ColumnInfo& ci = m_columns.at(i);
				const ColumnInfo& c2 = columns.at(ci.GetId());
                BASSERT(c2.GetFieldBytes() == ci.GetFieldBytes() && c2.GetId() == ci.GetId());
				memcpy(ptr, (const char*)record + c2.GetOffset(), ci.GetFieldBytes());
				ptr += ci.GetFieldBytes();
			}

			// 使用二分法在排序记录中找应该将record插入的位置
			const int CUTOFF = 3;
			int start = 0, end = m_record_count - 1;
			Range  range = { 0 };
			while (start + CUTOFF < end)
			{
				int mid = (start + end) / 2;
				RecordId midrid = *(int*)(m_buffer + mid*m_record_length);
				if (mid == midrid)
					return range;
				int cmp = CompareSortRecord(sort_record, m_buffer + mid*m_record_length);
				if (cmp <= 0)
					end = mid;
				else if (cmp > 0)
					start = mid;
			}
			range.start = start*m_extract_gap;
			range.end = end*m_extract_gap;
			return range;
		}

		void SortBuffer::ConvertToIndexing()
		{
            // 断言 m_extract_gap 是2次幂
            BASSERT(m_extract_gap >= 1);
			if (m_extract_gap != 1)
			{
				int ncount = m_record_count / m_extract_gap;
				for (int i = 0; i < ncount; i++)
				{
					char* src = m_buffer + (i*m_extract_gap + m_extract_gap_offset)*m_record_length;
					char* dst = m_buffer + i*m_record_length;
					if (dst != src)
						memcpy(dst, src, m_record_length);
				}
				m_record_length = ncount;
				m_issorting = false;
			}
		}

		int SortBuffer::CompareSortRecord(const void* pSortData1, const void* pSortData2)
		{
			for (int i = 0; i < (int)m_columns.size(); i++)
			{
				ColumnInfo& ci = m_columns[i];
				int ret = ci.m_comp((const char*)pSortData1 + ci.m_offset, (const char*)pSortData2 + ci.m_offset);
				if (ret != 0)
					return ci.m_dec ? -ret : ret;
			}
			return 0;
		}
	}//namespace ht
}//namespace qb
